#!/usr/bin/env python
# -*- coding:utf-8 -*-

import os, os.path
import re
import sys
import socket
import base64
import time

from jujuc import log
from jujuc.packet import Packet, enpack

from map_page import MapPage

def parseMaps(maps) :
    if not maps: return

    class   Result(object) : pass
    result = Result()
    result.error = None
    result.peers = []

    fields = maps.split('\t')
    for f in fields :
        k, v = f.split(':', 1)
        if 'key' == k :
            result.key = v
        elif 'status' == k :
            result.status = v
        elif 'error' == k :
            result.error = v
        elif 'cid' == k :
            result.cid = int(v)
        elif 'version' == k :
            result.version = v
        elif 'group' == k :
            result.group = v
        elif 'peer' == k :
            result.peers.append(v)

    return result
    # parseMaps()

class   Distributer(object) :
    def __init__(self, mapper, transport) :
        self.mapper = mapper
        self.transport = transport

        #group20@221.194.137.54:4705
        self.__pattern = re.compile( r'\s*(\w+)\@([^:]+):(\d+)\s*' )
        # __init__()

    def distribute(self, map_service, key, data) :
        maps = self.mapper.map(map_service, key)
        maps = parseMaps(maps)
        if maps.error :
            raise RuntimeError, 'MapError: map=%s key=%s' % (map_service, key)
        if len(maps.peers) < 1 :
            raise RuntimeError, 'No peer address available: map=%s key=%s' % (map_service, key)

        # this may optimized with a LRU cache
        __m = self.__pattern.search( maps.peers[0] )
        service = __m.group(1)
        host    = __m.group(2)
        port    = int(__m.group(3))

        tube = self.transport.getTCPTube(host, port)
        if tube is None :
            raise RuntimeError, 'No tube for "%s"' % maps.peers[0]

        packet = Packet()
        packet.setRequest()
        packet.service = service
        packet.body = base64.b64encode('\t'.join((str(maps.cid), data)))
        record, packid = enpack(packet)
        ok = tube.sendRecord(record)
        return ok
        # distribute()
    # class Distributer

#######################################################################

'''
if os.name == 'nt':
    # no such thing as WSAEPERM or error code 10001 according to winsock.h or MSDN
    EPERM = object()
    from errno import WSAEINVAL as EINVAL
    from errno import WSAEWOULDBLOCK as EWOULDBLOCK
    from errno import WSAEINPROGRESS as EINPROGRESS
    from errno import WSAEALREADY as EALREADY
    from errno import WSAECONNRESET as ECONNRESET
    from errno import WSAEISCONN as EISCONN
    from errno import WSAENOTCONN as ENOTCONN
    from errno import WSAEINTR as EINTR
    from errno import WSAENOBUFS as ENOBUFS
    from errno import WSAEMFILE as EMFILE
    # No such thing as WSAENFILE, either.
    ENFILE = object()
    # Nor ENOMEM
    ENOMEM = object()
    EAGAIN = EWOULDBLOCK
    from errno import WSAECONNRESET as ECONNABORTED
else:
    from errno import EPERM
    from errno import EINVAL
    from errno import EWOULDBLOCK
    from errno import EINPROGRESS
    from errno import EALREADY
    from errno import ECONNRESET
    from errno import EISCONN
    from errno import ENOTCONN
    from errno import EINTR
    from errno import ENOBUFS
    from errno import EMFILE
    from errno import ENFILE
    from errno import ENOMEM
    from errno import EAGAIN
    from errno import ECONNABORTED
'''
from errno import EALREADY, EINPROGRESS, EWOULDBLOCK, ECONNRESET, \
     ECONNREFUSED, ENOTCONN, ESHUTDOWN, EINTR, EISCONN, errorcode

class   TCPChannel(object) :
    def __init__(self, address) :
        self.address = address
        self.socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        #s.setblocking(0)
        #if fcntl and hasattr(fcntl, 'FD_CLOEXEC'):
        #    old = fcntl.fcntl(s.fileno(), fcntl.F_GETFD)
        #    fcntl.fcntl(s.fileno(), fcntl.F_SETFD, old | fcntl.FD_CLOEXEC)
        if 'posix' == os.name :
            self.socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

        self.connected = False
        self.terminator = '\n'

        self.reconnect()
        # __init__()

    def __del__(self) :
        self.handleClose()
        # __del__()

    def handleClose(self) :
        if self.socket :
            self.socket.close()
            self.connected = False
        # handleClose()

    def reconnect(self) :
        if self.connected : return

        errno = self.socket.connect_ex(self.address)
        if errno in (0, EISCONN) :
            self.connected = True
        elif errno in (EINPROGRESS, EALREADY, EWOULDBLOCK):
            self.connected = False
        else :
            self.connected = False
            log.error('fail to connect to "%s" -- [%d] %s', self.address, errno, errorcode[errno])
            #raise socket.error, (errno, errorcode[errno])
        # reconnect()

    def sendRecord(self, record) :
        out_buf = record + self.terminator
        if not self.connected :
            self.reconnect()

        while self.connected :
            if len(out_buf) < 1 :
                return True

            try :
                num_sent = self.socket.send(out_buf)
                if num_sent :
                    out_buf = out_buf[ num_sent: ]
            except socket.error, why :
                #self.handle_error()
                if why[0] == EWOULDBLOCK:
                    continue
                else :
                    log.error('fail to send record to "%s" -- %s', self.address, why[1])
                    raise
                return  False

        if not self.connected :
            return False
        return  True
        # sendRecord()

    def recvRecord(self) :
        if not self.connected :
            self.reconnect()

        while self.connected :
            try:
                data = self.socket.recv (self.ac_in_buffer_size)
            except socket.error, why:
                # winsock sometimes throws ENOTCONN
                if why[0] in [ECONNREFUSED, ECONNRESET, ENOTCONN, ESHUTDOWN]:
                    self.handleClose()
                    return None
                else:
                    raise
                ##self.handle_error()
                ##return None

            self.ac_in_buffer = self.ac_in_buffer + data

            lb = len(self.ac_in_buffer)
            terminator = self.terminator        ##self.getTerminator()
            terminator_len = len(terminator)
            index = self.ac_in_buffer.find(terminator)
            if index != -1:
                # we found the terminator
                if index > 0:
                    # don't bother reporting the empty string (source of subtle bugs)
                    record = self.ac_in_buffer[:index]
                else :
                    record = None
                self.ac_in_buffer = self.ac_in_buffer[index+terminator_len:]

                if record :
                    return record

        return None
        # recvRecord()
    # class TCPChannel

from persistentqueue import PersistentQueue
from persistentqueue import Empty as QueueEmpty

class   TCPTube(object) :
    def __init__(self, address, channel, cache_dir, cache_chunk_size=32768) :
        self.address = address
        self.channel = channel
        self.cache_dir = cache_dir
        if not os.path.exists(self.cache_dir) :
            os.makedirs(self.cache_dir)
        self.cache_chunk_size = cache_chunk_size
        self.cache = None

        # initialize cache
        self.cache = PersistentQueue(self.cache_dir, self.cache_chunk_size)
        if len(self.cache) > 0 :
            self.record_sending = self.cache.first()
        else :
            self.record_sending = None
        # __init__()

    def __del__(self) :
        if self.cache :
            self.cache.sync()
            self.cache.close()
            self.cache = None
        # __del__()

    def sendRecord(self, record) :
        '''

        return 1 -- queued, 0 -- send-out, -1 -- send-fail
        '''
        if self.record_sending :
            # more records to send, queue it
            self.cache.push(record)
            queued = True
        else :
            self.record_sending = record
            queued = False

        while True :
            #log.debug('#### %d items cached.', len(self.cache))
            #log.debug('sending record: "%s"', self.record_sending)
            ok = self.channel.sendRecord(self.record_sending)
            if ok :
                self.record_sending = None

                if len(self.cache) > 0 :
                    self.cache.pop()
                    try :
                        self.record_sending = self.cache.first()
                    except QueueEmpty, why :
                        # send out all cached record
                        self.cache.sync()
                        return 0
                    except :
                        nil, t, v, tbinfo = compact_traceback()
                        log.error('fail to pop from tube to %s -- (%s:%s %s)', address, t, v, tbinfo)
                        raise
                else :
                    # send out all cached record
                    return 0
            else :
                if not queued :
                    try :
                        self.cache.push(record)
                    except :
                        nil, t, v, tbinfo = compact_traceback()
                        log.error('fail to push to tube to %s -- (%s:%s %s)', address, t, v, tbinfo)
                        raise
                return 1
        # send()

    def recvRecord(self) :
        return self.channel.recvRecord()
        # recv()

    def flushSendingRecord(self) :
        '''
        return True on flush done, False on flush need
        '''
        if self.record_sending is None :
            return True

        while True :
            #log.debug('#### %d items cached.', len(self.cache))
            #log.debug('sending record: "%s"', self.record_sending)
            ok = self.channel.sendRecord(self.record_sending)
            if ok :
                self.record_sending = None

                if len(self.cache) > 0 :
                    self.cache.pop()
                    try :
                        self.record_sending = self.cache.first()
                    except QueueEmpty, why :
                        # send out all cached record
                        self.cache.sync()
                        return True
                    except :
                        nil, t, v, tbinfo = compact_traceback()
                        log.error('fail to pop from tube to %s -- (%s:%s %s)', address, t, v, tbinfo)
                        return False
                else :
                    # send out all cached record
                    return True
            else :
                return False
        # flushSendingRecord()

    # class TCPTube

class   Transport(object) :
    def __init__(self, cache_dir, cache_chunk_size) :
        self.cache_dir = cache_dir
        self.cache_chunk_size = cache_chunk_size

        self.channels = {}
        self.tubes = {}
        # __init__()

    def __del__(self) :
        for addr, tube in self.tubes.iteritems() :
            del tube
        for addr, channel in self.channels.iteritems() :
            del channel
        # __del__()

    def getTCPChannel(self, host, port) :
        address = (host, port)
        if address in self.channels :
            return self.channels[ address ]

        channel = TCPChannel( address )
        self.channels[ address ] = channel
        return channel
        # getTCPChannel()

    def getTCPTube(self, host, port) :
        address = (host, port)
        if address in self.tubes :
            return self.tubes[ address ]

        try :
            channel = self.getTCPChannel(host, port)
        except :
            nil, t, v, tbinfo = compact_traceback()
            log.error('fail to get channel to (%s:%d) -- (%s:%s %s)', host, port, t, v, tbinfo)
            return None

        tube_cache_dir = ''.join( (self.cache_dir, '/', '%s_%d' % (host, port)) )
        tube_cache_chunk_size = self.cache_chunk_size
        tube = TCPTube(address, channel, tube_cache_dir, tube_cache_chunk_size)
        self.tubes[ address ] = tube
        return tube
        # getTCPTube()

    def flushTCPTube(self) :
        '''
        flush all records cached in tube

        return 1 -- need flush again, 0 -- all flushed
        return False -- need flush again, True -- all flushed
        '''
        all_flushed = True
        for addr,tube in self.tubes.iteritems() :
            ok = tube.flushSendingRecord()
            if not ok :
                all_flushed = False

        if all_flushed :
            return True
        else :
            return False
        # flushTCPTube()

    # class Transport

#######################################################################

def distributeFile(distributer, mapservice, datafile, keyfield) :
    kn = keyfield - 1

    line_count = 0
    line_slice = 10000
    time_start = 0
    time_stop = 1

    for line in file(datafile) :
        line = line.strip()
        fields = line.split()
        if len(fields) <= kn :
            # invalid line
            continue
        key = fields[kn]

        if line_count == 0 :
            time_start = time.time()
        line_count += 1
        if (line_count >= line_slice ) :
            time_stop = time.time()
            time_last = time_stop - time_start
            tps = line_slice / time_last
            log.info('lines:%d time-last-in-seconds:%.2f tps:%.2f', line_slice, time_last, tps)
            line_count = 0

        try :
            ok = distributer.distribute(mapservice, key, line)
        except :
            nil, t, v, tbinfo = compact_traceback()
            log.error('fail to distribute: "%s" -- (%s:%s %s)', line, t, v, tbinfo)
            # go on to distribute next record
            continue

        if ok < 0 :
            log.error('FAIL %s %s %s', mapservice, key, line)
        #elif ok > 0 :
        #    log.error('CACHE %s %s %s', mapservice, line)
    # distributeFile

def compact_traceback():
    t, v, tb = sys.exc_info()
    tbinfo = []
    assert tb # Must have a traceback
    while tb:
        tbinfo.append((
            tb.tb_frame.f_code.co_filename,
            tb.tb_frame.f_code.co_name,
            str(tb.tb_lineno)
            ))
        tb = tb.tb_next

    # just to be safe
    del tb

    file, function, line = tbinfo[-1]
    info = ' '.join(['[%s|%s|%s]' % x for x in tbinfo])
    return (file, function, line), t, v, info


#######################################################################

import signal

will_stop = False

def signal_stop( signo, frame ) :
    global  will_stop
    log.debug( "JUJUB catch #%d signal, would stop.", signo )
    will_stop = True
    return
    # signal_stop

def main() :
    from optparse import OptionParser
    parser = OptionParser()
    parser.add_option("-v", "--verbose", dest="verbose", action="store_true", default=False,
                        help='show detail information')
    parser.add_option("", "--map-conf", dest="mapconf", type="string",
                        help='configure for map')

    parser.add_option("", "--map-service", dest="mapservice", type="string",
                        help='service of map to distribute data')
    parser.add_option("", "--data-file", dest="datafile", type="string",
                        help='datafile to be distribute')
    parser.add_option("", "--key-field", dest="keyfield", type="int", default=1,
                        help='which field is key data (1-based)')

    parser.add_option("", "--cache-dir", dest="cache_dir", type="string",
                        help='where cache files locate')
    parser.add_option("", "--cache-chunk-size", dest="cache_chunk_size", type="int", default=8192,
                        help='how many records each cache chunk can hold')

    parser.add_option("", "--log-dir", dest="log_dir", type="string",
                        help='where cache files locate')

    (options, args) = parser.parse_args()
    if options.mapconf is None or     \
       options.datafile is None or    \
       options.mapservice is None or  \
       options.cache_dir is None :
        parser.print_help( sys.stdout )
        return

    if options.log_dir :
        log.setLogPath( options.log_dir )
    if options.verbose :
        log.enableDebug()
    else :
        log.disableDebug()

    signal.signal( signal.SIGTERM, signal_stop )
    if 'posix' == os.name :
        signal.signal( signal.SIGQUIT, signal_stop )

    mapper = MapPage()
    mapper.open(options.mapconf)

    transport = Transport(options.cache_dir, options.cache_chunk_size)

    distributer = Distributer(mapper, transport)

    distributeFile(distributer, options.mapservice, options.datafile, options.keyfield)

    global  will_stop
    try :
        while not will_stop :
            ok = transport.flushTCPTube()
            if ok :
                break

            time.sleep(1)
    except KeyboardInterrupt :
        pass
    except :
        nil, t, v, tbinfo = compact_traceback()
        log.error('flush interruptted -- (%s:%s %s)', t, v, tbinfo)

    mapper.close()

    # main()

if  __name__ == '__main__' :
    main()

    import sys
    sys.exit(0)
